package cn.tedu.micro.demo.client;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;

/**
 * 测试生产发送确认机制的案例
 */
@Slf4j
public class RabbitmqPublishConfirm {
    //向queue01里发送一条消息
    @Test
    public void syncConfirm() throws IOException, TimeoutException, InterruptedException {
        //1.获取连接
        Channel channel = RabbitChannelManager.newChannel();
        //2.组织消息
        byte[] body = "hello rabbitmq".getBytes(StandardCharsets.UTF_8);
        String routingKey="queue01";
        //3.调用api发送消息
        //3.1开启确认模式
        channel.confirmSelect();
        long start=System.currentTimeMillis();
        channel.basicPublish("",routingKey,null,body);
        //3.2让线程等待消息发送反馈 timeout 5秒就算超时
        boolean ack = channel.waitForConfirms(5000);
        if (ack){
            log.info("消息发送成功");
        }else{
            log.info("消息发送失败");
        }
        long end=System.currentTimeMillis();
        log.info("耗时:{}MS",(end-start));
        //4.使用完毕 关闭连接
        channel.close();
    }

    @Test
    public void asyncConfirm() throws IOException, TimeoutException, InterruptedException {
        //1.获取连接
        Channel channel = RabbitChannelManager.newChannel();
        //2.组织消息
        byte[] body = "hello rabbitmq".getBytes(StandardCharsets.UTF_8);
        String routingKey="queue01";
        //3.调用api发送消息
        //3.1开启确认模式
        channel.confirmSelect();
        //3.2添加一个监听对象
        channel.addConfirmListener(new ConfirmListener() {
            //发送反馈成功
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                log.info("发送成功");
            }
            //发送反馈失败
            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                log.info("发送失败");
            }
        });
        long start=System.currentTimeMillis();
        channel.basicPublish("",routingKey,null,body);
        long end=System.currentTimeMillis();
        log.info("耗时:{}MS",(end-start));
        //4.使用完毕 关闭连接
        channel.close();
    }
    @Test
    public void txConfirm() throws IOException, TimeoutException, InterruptedException {
        //1.获取连接
        Channel channel = RabbitChannelManager.newChannel();
        //2.组织消息
        byte[] body = "hello rabbitmq".getBytes(StandardCharsets.UTF_8);
        String routingKey="queue01";
        //3.调用api发送消息
        //3.1开启事务确认
        channel.txSelect();
        long start = System.currentTimeMillis();
        try {
            //3.2发送事务消息: 主要目的就是检查rabbitmq是否可连接
            channel.basicPublish("", routingKey, null, body);
            //3.3执行其他业务步骤
            log.info("把当前订单提交了");
            //没有异常或者执行反馈true 提交消息
            channel.txCommit();
        }catch (Exception e){
            //有异常或者执行反馈false 回滚消息
            channel.txRollback();
        }
        long end=System.currentTimeMillis();
        log.info("耗时:{}MS",(end-start));
        //4.使用完毕 关闭连接
        channel.close();
    }
}
